package cn.doitedu.rtdw.data_dashboard;

import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Dash03_OrderTypeSum {

    public static void main(String[] args) {
        // 创建编程入口
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:/d:/ckpt");
        env.getCheckpointConfig().setTolerableCheckpointFailureNumber(3);
        env.setParallelism(1);
        // 状态后端
        env.setStateBackend(new HashMapStateBackend());

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        // 创建cdc连接器表，映射MySQL中视频信息表： cms_video
        tenv.executeSql(
                "CREATE TABLE order_test_sum (      " +
                        "      id BIGINT,              " +
                        "      s1 INT,                 " +
                        "      s2 INT,                 " +
                        "      s3 INT,                 " +
                        "      pt as proctime() ,                      " +
                        "     PRIMARY KEY (id) NOT ENFORCED            " +
                        "     ) WITH (                                 " +
                        "     'connector' = 'mysql-cdc',               " +
                        "     'hostname' = 'doitedu'   ,               " +
                        "     'port' = '3306'          ,               " +
                        "     'username' = 'root'      ,               " +
                        "     'password' = 'root'      ,               " +
                        "     'database-name' = 'rtmk',                " +
                        "     'table-name' = 'order_test_sum'          " +
                        ")"
        );

        tenv.executeSql(" SELECT\n" +
                "\n" +
                "   sum(if(s1=1 and s2=1,1,0)) as zb1,\n" +
                "   sum(if(s1=2 and s3=2,1,0)) as zb2,\n" +
                "   sum(if(s2=1 and s3=1,1,0)) as zb3\n" +
                "\n" +
                "FROM  \n" +
                "TABLE(\n" +
                "  TUMBLE(TABLE order_test_sum,DESCRIPTOR(pt),INTERVAL '5' SECOND)\n" +
                ")\n" +
                "GROUP BY \n" +
                "  window_start,\n" +
                "  window_end  ").print();



    }

}
